cc @nodejs/streams

@mcollina I quickly lookup and only we need destroy remaining readable streams if it is an Array when error occurs on previous readable readings. This control not necessary for iterables because stream will initialize lazy. We do not have control of writable stream in this scenario. Following solutions can solve the problem.

async function * streamSeriesIterator(streams) {
  let index = 0;

  try {
    for await (const stream of streams) {
      index++;
      yield* stream;
    }
  } finally {
    if (streams instanceof Array) {
      streams.slice(index).forEach(stream => stream.destroy());
    }
  }
}

or

async function * streamSeriesIterator(streams) {
  let index = 0;

  try {
    for await (const stream of streams) {
      index++;
      yield* stream;
    }
  } catch (err) {
    if (streams instanceof Array) {
      streams.slice(index).forEach(stream => stream.destroy(err));
    }
    
    throw err;
  }
}

As an alternative suggestion... consider adding this utility to the streams/consumers module and making it work with any stream.Readable, ReadableStream, and async iterable...

import { concat, arrayBuffer } from 'node:stream/consumers';

const readable1 = new ReadableStream();
const readable2 = new ReadableStream();
const readable3 = new stream.Readable();

for await (const chunk of concat(readable1, readable2, readable3)) {
  console.log(chunk);
}

// Let's say you want all of the contents in a single ArrayBuffer...

await arrayBuffer(concat(readable1, readable2, readable3));

The return value would be just an async iterator. If someone wanted a stream.Readable from the results, it would be as simple as... stream.Readable.from(concat(readable1, readable2, readable3)) ... getting a ReadableStream from it is a bit more complicated but also straightforward...

const readable = new ReadableStream({
  async pull(c) {
    for await (const chunk of concat(readable1, readable2, readable3))
      c.enqueue(chunk);
    c.close();
  }
});

@tugrul unfortunately #40075 (comment) does not work for node streams. They can error at any time even if they are not being consumed, you need to add on('error') handlers.

Hello, I'm doing some backlog hygiene on all PRs related to networking and streaming modules. There hasn't been any activity here for quite a while so I'm going to mark it stalled. It will auto-close in ~30 days. This is not a rejection; if you've like to pick it back up, rebase onto latest main and give me a ping.

Thank you!

This issue/PR was marked as stalled, it will be automatically closed in 30 days. If it should remain open, please leave a comment explaining why it should remain open.

Codecov Report

❌ Patch coverage is 93.73434% with 25 lines in your changes missing coverage. Please review.
✅ Project coverage is 90.14%. Comparing base (4a5eb1c) to head (d019ad6).
⚠️ Report is 9 commits behind head on main.

@@            Coverage Diff             @@
##             main   #40075      +/-   ##
==========================================
- Coverage   90.15%   90.14%   -0.02%     
==========================================
  Files         744      745       +1     
  Lines      242517   242917     +400     
  Branches    45688    45782      +94     
==========================================
+ Hits       218642   218968     +326     
- Misses      15358    15431      +73     
- Partials     8517     8518       +1